package com.sunzm.flink.datastream.scala

import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala._


/**
 *
 * 从sockert中获取数据，计算wordcount - scala 版
 *
 * @author Administrator
 * @version 1.0
 * @date 2021-06-16 16:35
 */
object SocketSourceScala {
  private val isLocal = true
  def main(args: Array[String]): Unit = {
    //1.创建执行的环境
    //1.创建执行的环境
    val env: StreamExecutionEnvironment = if (isLocal) {
      StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    } else {
      StreamExecutionEnvironment.getExecutionEnvironment
    }

    //2.从socket中获取数据
    val socketDataStream: DataStream[String] = env.socketTextStream("82.156.210.70", 9999)

    //3.计算wordcount
    val wordCountDataStream = socketDataStream.flatMap(_.split(","))
      .map((_, 1))
      .keyBy(_._1)
      .sum(1)

    //4. 打印结果
    wordCountDataStream.print()

    //5.执行
    env.execute(this.getClass.getSimpleName.stripSuffix("$"))
  }
}
